/*-------------------------------------------------------------------------
 *
 * subscriptioncmds.c
 *		订阅目录操作函数
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *		src/backend/commands/subscriptioncmds.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/htup_details.h"
#include "access/table.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/objectaddress.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "commands/defrem.h"
#include "commands/event_trigger.h"
#include "commands/subscriptioncmds.h"
#include "executor/executor.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "pgstat.h"
#include "replication/logicallauncher.h"
#include "replication/origin.h"
#include "replication/slot.h"
#include "replication/walreceiver.h"
#include "replication/walsender.h"
#include "replication/worker_internal.h"
#include "storage/lmgr.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/guc.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/syscache.h"

/*
 * 用户在CREATE/ALTER SUBSCRIPTION命令中可以指定的选项。
 */
#define SUBOPT_CONNECT				0x00000001
#define SUBOPT_ENABLED				0x00000002
#define SUBOPT_CREATE_SLOT			0x00000004
#define SUBOPT_SLOT_NAME			0x00000008
#define SUBOPT_COPY_DATA			0x00000010
#define SUBOPT_SYNCHRONOUS_COMMIT	0x00000020
#define SUBOPT_REFRESH				0x00000040
#define SUBOPT_BINARY				0x00000080
#define SUBOPT_STREAMING			0x00000100
#define SUBOPT_TWOPHASE_COMMIT		0x00000200
#define SUBOPT_DISABLE_ON_ERR		0x00000400
#define SUBOPT_LSN					0x00000800
#ifdef FDD
#define SUBOPT_FAILOVER  			0x00010000
#endif


/* 检查'val'是否设置了'bits' */
#define IsSet(val, bits)  (((val) & (bits)) == (bits))

/*
 * 用于保存表示用户提供的CREATE/ALTER
 * SUBSCRIPTION命令选项及其解析/默认值的位图的结构。
 */
typedef struct SubOpts
{
	bits32		specified_opts;
	char	   *slot_name;
	char	   *synchronous_commit;
	bool		connect;
	bool		enabled;
	bool		create_slot;
	bool		copy_data;
	bool		refresh;
	bool		binary;
	bool		streaming;
	bool		twophase;
	bool		disableonerr;
#ifdef FDD
	bool		failover;
#endif
	XLogRecPtr	lsn;
} SubOpts;

static List *fc_fetch_table_list(WalReceiverConn *fc_wrconn, List *fc_publications);
static void fc_check_duplicates_in_publist(List *fc_publist, Datum *fc_datums);
static List *fc_merge_publications(List *fc_oldpublist, List *fc_newpublist, bool fc_addpub, const char *fc_subname);
static void fc_ReportSlotConnectionError(List *fc_rstates, Oid fc_subid, char *fc_slotname, char *fc_err);


/*
 * CREATE和ALTER SUBSCRIPTION命令的通用选项解析函数。
 *
 * 由于并不是所有选项在这两个命令中都可以指定，如果指定了相互排斥的选项，
 * 此函数将报告错误。
 */
static void fc_parse_subscription_options(ParseState *fc_pstate, List *fc_stmt_options,
						   bits32 fc_supported_opts, SubOpts *fc_opts)
{
	ListCell   *fc_lc;

	/* 初始时清除选项。 */
	memset(fc_opts, 0, sizeof(SubOpts));

	/* 调用者必须预期某些选项 */
	Assert(fc_supported_opts != 0);

	/* 如果支持连接选项，则这些其他选项也需要被支持。 */
	Assert(!IsSet(fc_supported_opts, SUBOPT_CONNECT) ||
		   IsSet(fc_supported_opts, SUBOPT_ENABLED | SUBOPT_CREATE_SLOT |
				 SUBOPT_COPY_DATA));

	/* 为布尔支持选项设置默认值。 */
	if (IsSet(fc_supported_opts, SUBOPT_CONNECT))
		fc_opts->connect = true;
	if (IsSet(fc_supported_opts, SUBOPT_ENABLED))
		fc_opts->enabled = true;
	if (IsSet(fc_supported_opts, SUBOPT_CREATE_SLOT))
		fc_opts->create_slot = true;
	if (IsSet(fc_supported_opts, SUBOPT_COPY_DATA))
		fc_opts->copy_data = true;
	if (IsSet(fc_supported_opts, SUBOPT_REFRESH))
		fc_opts->refresh = true;
	if (IsSet(fc_supported_opts, SUBOPT_BINARY))
		fc_opts->binary = false;
	if (IsSet(fc_supported_opts, SUBOPT_STREAMING))
		fc_opts->streaming = false;
	if (IsSet(fc_supported_opts, SUBOPT_TWOPHASE_COMMIT))
		fc_opts->twophase = false;
	if (IsSet(fc_supported_opts, SUBOPT_DISABLE_ON_ERR))
		fc_opts->disableonerr = false;
#ifdef FDD
	if (IsSet(fc_supported_opts, SUBOPT_FAILOVER))
			fc_opts->failover = false;
#endif


	/* 解析选项 */
	foreach(fc_lc, fc_stmt_options)
	{
		DefElem    *fc_defel = (DefElem *) lfirst(fc_lc);

		if (IsSet(fc_supported_opts, SUBOPT_CONNECT) &&
			strcmp(fc_defel->defname, "connect") == 0)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_CONNECT))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_CONNECT;
			fc_opts->connect = defGetBoolean(fc_defel);
		}
		else if (IsSet(fc_supported_opts, SUBOPT_ENABLED) &&
				 strcmp(fc_defel->defname, "enabled") == 0)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_ENABLED))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_ENABLED;
			fc_opts->enabled = defGetBoolean(fc_defel);
		}
		else if (IsSet(fc_supported_opts, SUBOPT_CREATE_SLOT) &&
				 strcmp(fc_defel->defname, "create_slot") == 0)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_CREATE_SLOT))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_CREATE_SLOT;
			fc_opts->create_slot = defGetBoolean(fc_defel);
		}
		else if (IsSet(fc_supported_opts, SUBOPT_SLOT_NAME) &&
				 strcmp(fc_defel->defname, "slot_name") == 0)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_SLOT_NAME))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_SLOT_NAME;
			fc_opts->slot_name = defGetString(fc_defel);

			/* 将slot_name设置为NONE被视为没有插槽名称。 */
			if (strcmp(fc_opts->slot_name, "none") == 0)
				fc_opts->slot_name = NULL;
			else
				ReplicationSlotValidateName(fc_opts->slot_name, ERROR);
		}
		else if (IsSet(fc_supported_opts, SUBOPT_COPY_DATA) &&
				 strcmp(fc_defel->defname, "copy_data") == 0)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_COPY_DATA))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_COPY_DATA;
			fc_opts->copy_data = defGetBoolean(fc_defel);
		}
		else if (IsSet(fc_supported_opts, SUBOPT_SYNCHRONOUS_COMMIT) &&
				 strcmp(fc_defel->defname, "synchronous_commit") == 0)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_SYNCHRONOUS_COMMIT))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_SYNCHRONOUS_COMMIT;
			fc_opts->synchronous_commit = defGetString(fc_defel);

			/* 测试给定值是否对于synchronous_commit GUC有效。 */
			(void) set_config_option("synchronous_commit", fc_opts->synchronous_commit,
									 PGC_BACKEND, PGC_S_TEST, GUC_ACTION_SET,
									 false, 0, false);
		}
		else if (IsSet(fc_supported_opts, SUBOPT_REFRESH) &&
				 strcmp(fc_defel->defname, "refresh") == 0)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_REFRESH))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_REFRESH;
			fc_opts->refresh = defGetBoolean(fc_defel);
		}
		else if (IsSet(fc_supported_opts, SUBOPT_BINARY) &&
				 strcmp(fc_defel->defname, "binary") == 0)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_BINARY))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_BINARY;
			fc_opts->binary = defGetBoolean(fc_defel);
		}
		else if (IsSet(fc_supported_opts, SUBOPT_STREAMING) &&
				 strcmp(fc_defel->defname, "streaming") == 0)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_STREAMING))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_STREAMING;
			fc_opts->streaming = defGetBoolean(fc_defel);
		}
		else if (strcmp(fc_defel->defname, "two_phase") == 0)
		{
			/*
			 * 不允许切换two_phase选项。这样做可能导致
			 * 事务缺失并导致不一致的副本。
			 * 请参阅worker.c顶部的注释
			 *
			 * 注意：不支持的twophase表明此调用来源于
			 * AlterSubscription。
			 */
			if (!IsSet(fc_supported_opts, SUBOPT_TWOPHASE_COMMIT))
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("unrecognized subscription parameter: \"%s\"", fc_defel->defname)));

			if (IsSet(fc_opts->specified_opts, SUBOPT_TWOPHASE_COMMIT))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_TWOPHASE_COMMIT;
			fc_opts->twophase = defGetBoolean(fc_defel);
		}
		else if (IsSet(fc_supported_opts, SUBOPT_DISABLE_ON_ERR) &&
				 strcmp(fc_defel->defname, "disable_on_error") == 0)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_DISABLE_ON_ERR))
				errorConflictingDefElem(fc_defel, fc_pstate);

			fc_opts->specified_opts |= SUBOPT_DISABLE_ON_ERR;
			fc_opts->disableonerr = defGetBoolean(fc_defel);
		}
		else if (IsSet(fc_supported_opts, SUBOPT_LSN) &&
				 strcmp(fc_defel->defname, "lsn") == 0)
		{
			char	   *fc_lsn_str = defGetString(fc_defel);
			XLogRecPtr	fc_lsn;

			if (IsSet(fc_opts->specified_opts, SUBOPT_LSN))
				errorConflictingDefElem(fc_defel, fc_pstate);

			/* 将lsn设置为NONE被视为重置LSN */
			if (strcmp(fc_lsn_str, "none") == 0)
				fc_lsn = InvalidXLogRecPtr;
			else
			{
				/* 将参数解析为LSN */
				fc_lsn = DatumGetLSN(DirectFunctionCall1(pg_lsn_in,
													  CStringGetDatum(fc_lsn_str)));

				if (XLogRecPtrIsInvalid(fc_lsn))
					ereport(ERROR,
							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
							 errmsg("invalid WAL location (LSN): %s", fc_lsn_str)));
			}

			fc_opts->specified_opts |= SUBOPT_LSN;
			fc_opts->lsn = fc_lsn;
		}
#ifdef FDD
		else if (IsSet(fc_supported_opts, SUBOPT_FAILOVER) && 
			strcmp(fc_defel->defname, "failover") == 0)
		{
			 if (!fc_opts->create_slot)
			 {
				 ereport(WARNING,
						 (errcode(ERRCODE_SYNTAX_ERROR),
						  errmsg("cannot change existing slot failover state")));
			 }
			 else
			 {				 
				 fc_opts->specified_opts |= SUBOPT_FAILOVER;
				 fc_opts->failover = defGetBoolean(fc_defel);
			 }
		}
#endif
		else
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("unrecognized subscription parameter: \"%s\"", fc_defel->defname)));
	}

	/*
	 * 我们已被明确要求不连接，这需要一些
	 * 附加处理。
	 */
	if (!fc_opts->connect && IsSet(fc_supported_opts, SUBOPT_CONNECT))
	{
		/* 检查用户的不兼容选项。 */
		if (fc_opts->enabled &&
			IsSet(fc_opts->specified_opts, SUBOPT_ENABLED))
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
			/*- 翻译者：两个%s都是"option = value"形式的字符串 */
					 errmsg("%s and %s are mutually exclusive options",
							"connect = false", "enabled = true")));

		if (fc_opts->create_slot &&
			IsSet(fc_opts->specified_opts, SUBOPT_CREATE_SLOT))
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("%s and %s are mutually exclusive options",
							"connect = false", "create_slot = true")));

		if (fc_opts->copy_data &&
			IsSet(fc_opts->specified_opts, SUBOPT_COPY_DATA))
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("%s and %s are mutually exclusive options",
							"connect = false", "copy_data = true")));

		/* 更改其他选项的默认值。 */
		fc_opts->enabled = false;
		fc_opts->create_slot = false;
		fc_opts->copy_data = false;
	}

	/*
	 * 当使用slot_name = NONE时，进行不允许组合的额外检查。
	 */
	if (!fc_opts->slot_name &&
		IsSet(fc_opts->specified_opts, SUBOPT_SLOT_NAME))
	{
		if (fc_opts->enabled)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_ENABLED))
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
				/*- 翻译者：两个%s都是"option = value"形式的字符串 */
						 errmsg("%s and %s are mutually exclusive options",
								"slot_name = NONE", "enabled = true")));
			else
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
				/*- 翻译者：两个%s都是"option = value"形式的字符串 */
						 errmsg("subscription with %s must also set %s",
								"slot_name = NONE", "enabled = false")));
		}

		if (fc_opts->create_slot)
		{
			if (IsSet(fc_opts->specified_opts, SUBOPT_CREATE_SLOT))
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
				/*- 翻译者：两个%s都是"option = value"形式的字符串 */
						 errmsg("%s and %s are mutually exclusive options",
								"slot_name = NONE", "create_slot = true")));
			else
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
				/*- 翻译者：两个%s都是"option = value"形式的字符串 */
						 errmsg("subscription with %s must also set %s",
								"slot_name = NONE", "create_slot = false")));
		}
	}
}

/*
 * 从列表中添加publication名称到字符串中。
 */
static void fc_get_publications_str(List *fc_publications, StringInfo fc_dest, bool fc_quote_literal)
{
	ListCell   *fc_lc;
	bool		fc_first = true;

	Assert(list_length(fc_publications) > 0);

	foreach(fc_lc, fc_publications)
	{
		char	   *fc_pubname = strVal(lfirst(fc_lc));

		if (fc_first)
			fc_first = false;
		else
			appendStringInfoString(fc_dest, ", ");

		if (fc_quote_literal)
			appendStringInfoString(fc_dest, quote_literal_cstr(fc_pubname));
		else
		{
			appendStringInfoChar(fc_dest, '"');
			appendStringInfoString(fc_dest, fc_pubname);
			appendStringInfoChar(fc_dest, '"');
		}
	}
}

/*
 * 检查指定的出版物是否在发布者上存在。
 */
static void fc_check_publications(WalReceiverConn *fc_wrconn, List *fc_publications)
{
	WalRcvExecResult *fc_res;
	StringInfo	fc_cmd;
	TupleTableSlot *fc_slot;
	List	   *fc_publicationsCopy = NIL;
	Oid			fc_tableRow[1] = {TEXTOID};

	fc_cmd = makeStringInfo();
	appendStringInfoString(fc_cmd, "SELECT t.pubname FROM\n"
						   " pg_catalog.pg_publication t WHERE\n"
						   " t.pubname IN (");
	fc_get_publications_str(fc_publications, fc_cmd, true);
	appendStringInfoChar(fc_cmd, ')');

	fc_res = walrcv_exec(fc_wrconn, fc_cmd->data, 1, fc_tableRow);
	pfree(fc_cmd->data);
	pfree(fc_cmd);

	if (fc_res->status != WALRCV_OK_TUPLES)
		ereport(ERROR,
				errmsg("could not receive list of publications from the publisher: %s",
					   fc_res->err));

	fc_publicationsCopy = list_copy(fc_publications);

	/* 处理出版物。 */
	fc_slot = MakeSingleTupleTableSlot(fc_res->tupledesc, &TTSOpsMinimalTuple);
	while (tuplestore_gettupleslot(fc_res->tuplestore, true, false, fc_slot))
	{
		char	   *fc_pubname;
		bool		fc_isnull;

		fc_pubname = TextDatumGetCString(slot_getattr(fc_slot, 1, &fc_isnull));
		Assert(!fc_isnull);

		/* 从列表中删除发布者中存在的出版物。 */
		fc_publicationsCopy = list_delete(fc_publicationsCopy, makeString(fc_pubname));
		ExecClearTuple(fc_slot);
	}

	ExecDropSingleTupleTableSlot(fc_slot);

	walrcv_clear_result(fc_res);

	if (list_length(fc_publicationsCopy))
	{
		/* 准备非存在出版物的列表以用于错误消息。 */
		StringInfo	fc_pubnames = makeStringInfo();

		fc_get_publications_str(fc_publicationsCopy, fc_pubnames, false);
		ereport(WARNING,
				errcode(ERRCODE_UNDEFINED_OBJECT),
				errmsg_plural("publication %s does not exist on the publisher",
							  "publications %s do not exist on the publisher",
							  list_length(fc_publicationsCopy),
							  fc_pubnames->data));
	}
}

/*
 * 辅助函数，用于从字符串节点列表构建文本数组。
 */
static Datum fc_publicationListToArray(List *fc_publist)
{
	ArrayType  *fc_arr;
	Datum	   *fc_datums;
	MemoryContext fc_memcxt;
	MemoryContext fc_oldcxt;

	/* 为临时分配创建内存上下文。 */
	fc_memcxt = AllocSetContextCreate(CurrentMemoryContext,
								   "publicationListToArray to array",
								   ALLOCSET_DEFAULT_SIZES);
	fc_oldcxt = MemoryContextSwitchTo(fc_memcxt);

	fc_datums = (Datum *) palloc(sizeof(Datum) * list_length(fc_publist));

	fc_check_duplicates_in_publist(fc_publist, fc_datums);

	MemoryContextSwitchTo(fc_oldcxt);

	fc_arr = construct_array(fc_datums, list_length(fc_publist),
						  TEXTOID, -1, false, TYPALIGN_INT);

	MemoryContextDelete(fc_memcxt);

	return PointerGetDatum(fc_arr);
}

/*
 * 创建新的订阅。
 */
ObjectAddress CreateSubscription(ParseState *fc_pstate, CreateSubscriptionStmt *fc_stmt,
				   bool fc_isTopLevel)
{
	Relation	fc_rel;
	ObjectAddress fc_myself;
	Oid			fc_subid;
	bool		fc_nulls[Natts_pg_subscription];
	Datum		fc_values[Natts_pg_subscription];
	Oid			fc_owner = GetUserId();
	HeapTuple	fc_tup;
	char	   *fc_conninfo;
	char		fc_originname[NAMEDATALEN];
	List	   *fc_publications;
	bits32		fc_supported_opts;
	SubOpts		fc_opts = {0};

	/*
	 * 解析和检查选项。
	 *
	 * 连接和出版物不应在此处指定。
	 */
	fc_supported_opts = (SUBOPT_CONNECT | SUBOPT_ENABLED | SUBOPT_CREATE_SLOT |
					  SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA |
					  SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
					  SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
#ifndef FDD
					  SUBOPT_DISABLE_ON_ERR);
#else
					  SUBOPT_DISABLE_ON_ERR | SUBOPT_FAILOVER);
#endif
	fc_parse_subscription_options(fc_pstate, fc_stmt->options, fc_supported_opts, &fc_opts);

	/*
	 * 由于创建复制插槽不是事务性的，回滚
	 * 事务会留下已创建的复制插槽。因此，我们不能在
	 * 事务块中运行CREATE SUBSCRIPTION，如果创建了
	 * 复制插槽。
	 */
	if (fc_opts.create_slot)
		PreventInTransactionBlock(fc_isTopLevel, "CREATE SUBSCRIPTION ... WITH (create_slot = true)");

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to create subscriptions")));

	/*
	 * 如果使用适当的开关构建，当订阅名称的
	 * 回归测试约定被违反时发出警告。
	 */
#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
	if (strncmp(fc_stmt->subname, "regress_", 8) != 0)
		elog(WARNING, "subscriptions created by regression test cases should have names starting with \"regress_\"");
#endif

	fc_rel = table_open(SubscriptionRelationId, RowExclusiveLock);

	/* 检查名称是否已使用 */
	fc_subid = GetSysCacheOid2(SUBSCRIPTIONNAME, Anum_pg_subscription_oid,
							MyDatabaseId, CStringGetDatum(fc_stmt->subname));
	if (OidIsValid(fc_subid))
	{
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_OBJECT),
				 errmsg("subscription \"%s\" already exists",
						fc_stmt->subname)));
	}

	if (!IsSet(fc_opts.specified_opts, SUBOPT_SLOT_NAME) &&
		fc_opts.slot_name == NULL)
		fc_opts.slot_name = fc_stmt->subname;

	/* 订阅的 synchronous_commit 默认值为关闭。 */
	if (fc_opts.synchronous_commit == NULL)
		fc_opts.synchronous_commit = "off";

	fc_conninfo = fc_stmt->conninfo;
	fc_publications = fc_stmt->publication;

	/* 加载提供 libpq 调用的库。 */
	load_file("libpqwalreceiver", false);

	/* 检查连接信息字符串。 */
	walrcv_check_conninfo(fc_conninfo);

	/* 一切正常，形成一个新元组。 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));

	fc_subid = GetNewOidWithIndex(fc_rel, SubscriptionObjectIndexId,
							   Anum_pg_subscription_oid);
	fc_values[Anum_pg_subscription_oid - 1] = ObjectIdGetDatum(fc_subid);
	fc_values[Anum_pg_subscription_subdbid - 1] = ObjectIdGetDatum(MyDatabaseId);
	fc_values[Anum_pg_subscription_subskiplsn - 1] = LSNGetDatum(InvalidXLogRecPtr);
	fc_values[Anum_pg_subscription_subname - 1] =
		DirectFunctionCall1(namein, CStringGetDatum(fc_stmt->subname));
	fc_values[Anum_pg_subscription_subowner - 1] = ObjectIdGetDatum(fc_owner);
	fc_values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(fc_opts.enabled);
	fc_values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(fc_opts.binary);
	fc_values[Anum_pg_subscription_substream - 1] = BoolGetDatum(fc_opts.streaming);
	fc_values[Anum_pg_subscription_subtwophasestate - 1] =
		CharGetDatum(fc_opts.twophase ?
					 LOGICALREP_TWOPHASE_STATE_PENDING :
					 LOGICALREP_TWOPHASE_STATE_DISABLED);
	fc_values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(fc_opts.disableonerr);
	fc_values[Anum_pg_subscription_subconninfo - 1] =
		CStringGetTextDatum(fc_conninfo);
	if (fc_opts.slot_name)
		fc_values[Anum_pg_subscription_subslotname - 1] =
			DirectFunctionCall1(namein, CStringGetDatum(fc_opts.slot_name));
	else
		fc_nulls[Anum_pg_subscription_subslotname - 1] = true;
	fc_values[Anum_pg_subscription_subsynccommit - 1] =
		CStringGetTextDatum(fc_opts.synchronous_commit);
	fc_values[Anum_pg_subscription_subpublications - 1] =
		fc_publicationListToArray(fc_publications);

	fc_tup = heap_form_tuple(RelationGetDescr(fc_rel), fc_values, fc_nulls);

	/* 将元组插入目录。 */
	CatalogTupleInsert(fc_rel, fc_tup);
	heap_freetuple(fc_tup);

	recordDependencyOnOwner(SubscriptionRelationId, fc_subid, fc_owner);

	snprintf(fc_originname, sizeof(fc_originname), "pg_%u", fc_subid);
	replorigin_create(fc_originname);

	/*
	 * 连接远程端以执行请求的命令并获取表
	 * 信息。
	 */
	if (fc_opts.connect)
	{
		char	   *fc_err;
		WalReceiverConn *fc_wrconn;
		List	   *fc_tables;
		ListCell   *fc_lc;
		char		fc_table_state;

		/* 尝试连接到发布者。 */
		fc_wrconn = walrcv_connect(fc_conninfo, true, fc_stmt->subname, &fc_err);
		if (!fc_wrconn)
			ereport(ERROR,
					(errcode(ERRCODE_CONNECTION_FAILURE),
					 errmsg("could not connect to the publisher: %s", fc_err)));

		PG_TRY();
		{
			fc_check_publications(fc_wrconn, fc_publications);

			/*
			 * 根据是否被要求进行数据复制来设置同步状态。
			 */
			fc_table_state = fc_opts.copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY;

			/*
			 * 从发布者获取表列表并构建本地表状态
			 * 信息。
			 */
			fc_tables = fc_fetch_table_list(fc_wrconn, fc_publications);
			foreach(fc_lc, fc_tables)
			{
				RangeVar   *fc_rv = (RangeVar *) lfirst(fc_lc);
				Oid			fc_relid;

				fc_relid = RangeVarGetRelid(fc_rv, AccessShareLock, false);

				/* 检查支持的 relkind。 */
				CheckSubscriptionRelkind(get_rel_relkind(fc_relid),
										 fc_rv->schemaname, fc_rv->relname);

				AddSubscriptionRelState(fc_subid, fc_relid, fc_table_state,
										InvalidXLogRecPtr);
			}

			/*
			 * 如果请求，创建订阅的永久槽。我们
			 * 不会将初始快照用于任何目的，因此无需
			 * 导出它。
			 */
			if (fc_opts.create_slot)
			{
				bool		fc_twophase_enabled = false;

				Assert(fc_opts.slot_name);

				/*
				 * 即使设置了 two_phase，也不要在启用
				 * two-phase 的情况下创建槽。一旦所有表
				 * 同步并准备就绪，将启用它。这可以避免诸如准备
				 * 事务被跳过的竞争条件，因为变更未被应用
				 * 是由于在 should_apply_changes_for_rel() 中的检查，
				 * 当对应的表的表同步正在进行时。请参见
				 * worker.c 上方的注释。
				 *
				 * 请注意，如果指定了表但 copy_data 为假
				 * 那么安全地 upfront 启用 two_phase 是安全的，因为
				 * 那些表已经初始处于 READY 状态。当
				 * 订阅没有表时，我们将 twophase 状态保持为
				 * PENDING，以允许 ALTER SUBSCRIPTION ... REFRESH
				 * PUBLICATION 的操作。
				 */
				if (fc_opts.twophase && !fc_opts.copy_data && fc_tables != NIL)
					fc_twophase_enabled = true;

#ifdef FDD
				walrcv_create_slot(fc_wrconn, fc_opts.slot_name, false, fc_twophase_enabled, fc_opts.failover,
								   CRS_NOEXPORT_SNAPSHOT, NULL);
#else
				walrcv_create_slot(fc_wrconn, fc_opts.slot_name, false, fc_twophase_enabled,
								   CRS_NOEXPORT_SNAPSHOT, NULL);
#endif

				if (fc_twophase_enabled)
					UpdateTwoPhaseState(fc_subid, LOGICALREP_TWOPHASE_STATE_ENABLED);

				ereport(NOTICE,
						(errmsg("created replication slot \"%s\" on publisher",
								fc_opts.slot_name)));
			}
		}
		PG_FINALLY();
		{
			walrcv_disconnect(fc_wrconn);
		}
		PG_END_TRY();
	}
	else
		ereport(WARNING,
		/* 翻译者：%s 是 SQL ALTER 语句 */
				(errmsg("tables were not subscribed, you will have to run %s to subscribe the tables",
						"ALTER SUBSCRIPTION ... REFRESH PUBLICATION")));

	table_close(fc_rel, RowExclusiveLock);

	pgstat_create_subscription(fc_subid);

	if (fc_opts.enabled)
		ApplyLauncherWakeupAtCommit();

	ObjectAddressSet(fc_myself, SubscriptionRelationId, fc_subid);

	InvokeObjectPostCreateHook(SubscriptionRelationId, fc_subid, 0);

	return fc_myself;
}

static void fc_AlterSubscription_refresh(Subscription *fc_sub, bool fc_copy_data,
						  List *fc_validate_publications)
{
	char	   *fc_err;
	List	   *fc_pubrel_names;
	List	   *fc_subrel_states;
	Oid		   *fc_subrel_local_oids;
	Oid		   *fc_pubrel_local_oids;
	ListCell   *fc_lc;
	int			fc_off;
	int			fc_remove_rel_len;
	Relation	fc_rel = NULL;
	typedef struct SubRemoveRels
	{
		Oid			relid;
		char		state;
	} SubRemoveRels;
	SubRemoveRels *fc_sub_remove_rels;
	WalReceiverConn *fc_wrconn;

	/* 加载提供 libpq 调用的库。 */
	load_file("libpqwalreceiver", false);

	/* 尝试连接到发布者。 */
	fc_wrconn = walrcv_connect(fc_sub->conninfo, true, fc_sub->name, &fc_err);
	if (!fc_wrconn)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("could not connect to the publisher: %s", fc_err)));

	PG_TRY();
	{
		if (fc_validate_publications)
			fc_check_publications(fc_wrconn, fc_validate_publications);

		/* 从发布者获取表列表。 */
		fc_pubrel_names = fc_fetch_table_list(fc_wrconn, fc_sub->publications);

		/* 获取本地表列表。 */
		fc_subrel_states = GetSubscriptionRelations(fc_sub->oid);

		/*
		 * 构建本地表 oid 的 qsorted 数组以加快查找。这可能
		 * 包含数据库中的所有表，因此查找的速度
		 * 是重要的。
		 */
		fc_subrel_local_oids = palloc(list_length(fc_subrel_states) * sizeof(Oid));
		fc_off = 0;
		foreach(fc_lc, fc_subrel_states)
		{
			SubscriptionRelState *fc_relstate = (SubscriptionRelState *) lfirst(fc_lc);

			fc_subrel_local_oids[fc_off++] = fc_relstate->relid;
		}
		qsort(fc_subrel_local_oids, list_length(fc_subrel_states),
			  sizeof(Oid), oid_cmp);

		/*
		 * 我们希望从订阅中删除的表，并删除与之相应的槽
		 * 和来源。
		 */
		fc_sub_remove_rels = palloc(list_length(fc_subrel_states) * sizeof(SubRemoveRels));

		/*
		 * 遍历远程表，尝试将它们与本地已知
		 * 表进行匹配。如果表在本地未知，则为
		 * 它创建一个新的状态。
		 *
		 * 还为下一步构建远程表的本地 oid 数组。
		 */
		fc_off = 0;
		fc_pubrel_local_oids = palloc(list_length(fc_pubrel_names) * sizeof(Oid));

		foreach(fc_lc, fc_pubrel_names)
		{
			RangeVar   *fc_rv = (RangeVar *) lfirst(fc_lc);
			Oid			fc_relid;

			fc_relid = RangeVarGetRelid(fc_rv, AccessShareLock, false);

			/* 检查支持的 relkind。 */
			CheckSubscriptionRelkind(get_rel_relkind(fc_relid),
									 fc_rv->schemaname, fc_rv->relname);

			fc_pubrel_local_oids[fc_off++] = fc_relid;

			if (!bsearch(&fc_relid, fc_subrel_local_oids,
						 list_length(fc_subrel_states), sizeof(Oid), oid_cmp))
			{
				AddSubscriptionRelState(fc_sub->oid, fc_relid,
										fc_copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
										InvalidXLogRecPtr);
				ereport(DEBUG1,
						(errmsg_internal("table \"%s.%s\" added to subscription \"%s\"",
										 fc_rv->schemaname, fc_rv->relname, fc_sub->name)));
			}
		}

		/*
		 * 接下来使用我们收集的上述数据删除我们不再关心的表的状态
		 */
		qsort(fc_pubrel_local_oids, list_length(fc_pubrel_names),
			  sizeof(Oid), oid_cmp);

		fc_remove_rel_len = 0;
		for (fc_off = 0; fc_off < list_length(fc_subrel_states); fc_off++)
		{
			Oid			fc_relid = fc_subrel_local_oids[fc_off];

			if (!bsearch(&fc_relid, fc_pubrel_local_oids,
						 list_length(fc_pubrel_names), sizeof(Oid), oid_cmp))
			{
				char		fc_state;
				XLogRecPtr	fc_statelsn;

				/*
				 * 使用AccessExclusiveLock锁定pg_subscription_rel，以
				 * 防止在应用工作者尝试删除这些表的同时重新启动工作者时发生竞争条件。
				 *
				 * 即使此特定关系的新工作者被重新启动，它也将无法
				 * 取得任何进展，因为我们在事务结束之前持有对
				 * subscription_rel的独占锁。它将简单地退出，因为
				 * 没有相应的关系条目。
				 *
				 * 此锁定还确保在我们完成此刷新操作之前，关系的状态不会
				 * 发生变化。
				 */
				if (!fc_rel)
					fc_rel = table_open(SubscriptionRelRelationId, AccessExclusiveLock);

				/* 最后已知的关系状态。 */
				fc_state = GetSubscriptionRelState(fc_sub->oid, fc_relid, &fc_statelsn);

				fc_sub_remove_rels[fc_remove_rel_len].relid = fc_relid;
				fc_sub_remove_rels[fc_remove_rel_len++].state = fc_state;

				RemoveSubscriptionRel(fc_sub->oid, fc_relid);

				logicalrep_worker_stop(fc_sub->oid, fc_relid);

				/*
				 * 对于READY状态，我们已经删除了tablesync origin。
				 */
				if (fc_state != SUBREL_STATE_READY)
				{
					char		fc_originname[NAMEDATALEN];

					/*
					 * 如果存在，删除tablesync的origin跟踪。
					 *
					 * 可能尚未为tablesync工作者创建origin，这可能发生在
					 * SUBREL_STATE_FINISHEDCOPY之前的状态。应用工作者也
					 * 可能会并发尝试删除origin，而此时origin可能已经被移除。
					 * 出于这些原因，传递missing_ok = true。
					 */
					ReplicationOriginNameForTablesync(fc_sub->oid, fc_relid, fc_originname,
													  sizeof(fc_originname));
					replorigin_drop_by_name(fc_originname, true, false);
				}

				ereport(DEBUG1,
						(errmsg_internal("table \"%s.%s\" removed from subscription \"%s\"",
										 get_namespace_name(get_rel_namespace(fc_relid)),
										 get_rel_name(fc_relid),
										 fc_sub->name)));
			}
		}

		/*
		 * 删除与已删除表关联的tablesync插槽。这必须在
		 * 最后进行，因为否则如果在执行数据库操作时发生
		 * 错误，我们将无法回滚已删除的插槽。
		 */
		for (fc_off = 0; fc_off < fc_remove_rel_len; fc_off++)
		{
			if (fc_sub_remove_rels[fc_off].state != SUBREL_STATE_READY &&
				fc_sub_remove_rels[fc_off].state != SUBREL_STATE_SYNCDONE)
			{
				char		fc_syncslotname[NAMEDATALEN] = {0};

				/*
				 * 对于READY/SYNCDONE状态，我们知道tablesync插槽已
				 * 被tablesync工作者删除。
				 *
				 * 对于其他状态，没有确定性，插槽可能尚不存在。
				 * 此外，如果在删除某些插槽后失败，下次再次
				 * 尝试删除已经删除的插槽将失败。出于这些原因，
				 * 我们允许drop时missing_ok = true。
				 */
				ReplicationSlotNameForTablesync(fc_sub->oid, fc_sub_remove_rels[fc_off].relid,
												fc_syncslotname, sizeof(fc_syncslotname));
				ReplicationSlotDropAtPubNode(fc_wrconn, fc_syncslotname, true);
			}
		}
	}
	PG_FINALLY();
	{
		walrcv_disconnect(fc_wrconn);
	}
	PG_END_TRY();

	if (fc_rel)
		table_close(fc_rel, NoLock);
}

/*
 * 修改现有的订阅。
 */
ObjectAddress AlterSubscription(ParseState *fc_pstate, AlterSubscriptionStmt *fc_stmt,
				  bool fc_isTopLevel)
{
	Relation	fc_rel;
	ObjectAddress fc_myself;
	bool		fc_nulls[Natts_pg_subscription];
	bool		fc_replaces[Natts_pg_subscription];
	Datum		fc_values[Natts_pg_subscription];
	HeapTuple	fc_tup;
	Oid			fc_subid;
	bool		fc_update_tuple = false;
	Subscription *fc_sub;
	Form_pg_subscription fc_form;
	bits32		fc_supported_opts;
	SubOpts		fc_opts = {0};

	fc_rel = table_open(SubscriptionRelationId, RowExclusiveLock);

	/* 获取现有元组。 */
	fc_tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, MyDatabaseId,
							  CStringGetDatum(fc_stmt->subname));

	if (!HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("subscription \"%s\" does not exist",
						fc_stmt->subname)));

	fc_form = (Form_pg_subscription) GETSTRUCT(fc_tup);
	fc_subid = fc_form->oid;

	/* 必须是所有者 */
	if (!pg_subscription_ownercheck(fc_subid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
					   fc_stmt->subname);

	fc_sub = GetSubscription(fc_subid, false);

	/* 锁定订阅，以便其他人无法对其执行任何操作。 */
	LockSharedObject(SubscriptionRelationId, fc_subid, 0, AccessExclusiveLock);

	/* 形成一个新元组。 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));
	memset(fc_replaces, false, sizeof(fc_replaces));

	switch (fc_stmt->kind)
	{
		case ALTER_SUBSCRIPTION_OPTIONS:
			{
				fc_supported_opts = (SUBOPT_SLOT_NAME |
								  SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
								  SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR);

				fc_parse_subscription_options(fc_pstate, fc_stmt->options,
										   fc_supported_opts, &fc_opts);

				if (IsSet(fc_opts.specified_opts, SUBOPT_SLOT_NAME))
				{
					/*
					 * 订阅必须被禁用以允许slot_name为
					 * 'none'，否则应用工作者将反复尝试
					 * 使用该slot_name流式传输数据，而该slot_name
					 * 既不存在于发布者上，也不允许用户创建它。
					 */
					if (fc_sub->enabled && !fc_opts.slot_name)
						ereport(ERROR,
								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
								 errmsg("cannot set %s for enabled subscription",
										"slot_name = NONE")));

					if (fc_opts.slot_name)
						fc_values[Anum_pg_subscription_subslotname - 1] =
							DirectFunctionCall1(namein, CStringGetDatum(fc_opts.slot_name));
					else
						fc_nulls[Anum_pg_subscription_subslotname - 1] = true;
					fc_replaces[Anum_pg_subscription_subslotname - 1] = true;
				}

				if (fc_opts.synchronous_commit)
				{
					fc_values[Anum_pg_subscription_subsynccommit - 1] =
						CStringGetTextDatum(fc_opts.synchronous_commit);
					fc_replaces[Anum_pg_subscription_subsynccommit - 1] = true;
				}

				if (IsSet(fc_opts.specified_opts, SUBOPT_BINARY))
				{
					fc_values[Anum_pg_subscription_subbinary - 1] =
						BoolGetDatum(fc_opts.binary);
					fc_replaces[Anum_pg_subscription_subbinary - 1] = true;
				}

				if (IsSet(fc_opts.specified_opts, SUBOPT_STREAMING))
				{
					fc_values[Anum_pg_subscription_substream - 1] =
						BoolGetDatum(fc_opts.streaming);
					fc_replaces[Anum_pg_subscription_substream - 1] = true;
				}

				if (IsSet(fc_opts.specified_opts, SUBOPT_DISABLE_ON_ERR))
				{
					fc_values[Anum_pg_subscription_subdisableonerr - 1]
						= BoolGetDatum(fc_opts.disableonerr);
					fc_replaces[Anum_pg_subscription_subdisableonerr - 1]
						= true;
				}

				fc_update_tuple = true;
				break;
			}

		case ALTER_SUBSCRIPTION_ENABLED:
			{
				fc_parse_subscription_options(fc_pstate, fc_stmt->options,
										   SUBOPT_ENABLED, &fc_opts);
				Assert(IsSet(fc_opts.specified_opts, SUBOPT_ENABLED));

				if (!fc_sub->slotname && fc_opts.enabled)
					ereport(ERROR,
							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
							 errmsg("cannot enable subscription that does not have a slot name")));

				fc_values[Anum_pg_subscription_subenabled - 1] =
					BoolGetDatum(fc_opts.enabled);
				fc_replaces[Anum_pg_subscription_subenabled - 1] = true;

				if (fc_opts.enabled)
					ApplyLauncherWakeupAtCommit();

				fc_update_tuple = true;
				break;
			}

		case ALTER_SUBSCRIPTION_CONNECTION:
			/* 加载提供 libpq 调用的库。 */
			load_file("libpqwalreceiver", false);
			/* 检查连接信息字符串。 */
			walrcv_check_conninfo(fc_stmt->conninfo);

			fc_values[Anum_pg_subscription_subconninfo - 1] =
				CStringGetTextDatum(fc_stmt->conninfo);
			fc_replaces[Anum_pg_subscription_subconninfo - 1] = true;
			fc_update_tuple = true;
			break;

		case ALTER_SUBSCRIPTION_SET_PUBLICATION:
			{
				fc_supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH;
				fc_parse_subscription_options(fc_pstate, fc_stmt->options,
										   fc_supported_opts, &fc_opts);

				fc_values[Anum_pg_subscription_subpublications - 1] =
					fc_publicationListToArray(fc_stmt->publication);
				fc_replaces[Anum_pg_subscription_subpublications - 1] = true;

				fc_update_tuple = true;

				/* 如果用户要求我们刷新，则刷新。 */
				if (fc_opts.refresh)
				{
					if (!fc_sub->enabled)
						ereport(ERROR,
								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
								 errmsg("ALTER SUBSCRIPTION with refresh is not allowed for disabled subscriptions"),
								 errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION ... WITH (refresh = false).")));

					/*
					 * 查看ALTER_SUBSCRIPTION_REFRESH以了解为什么
					 * 这不被允许的详细信息。
					 */
					if (fc_sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && fc_opts.copy_data)
						ereport(ERROR,
								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
								 errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
								 errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));

					PreventInTransactionBlock(fc_isTopLevel, "ALTER SUBSCRIPTION with refresh");

					/* 确保刷新看到新的出版物列表。 */
					fc_sub->publications = fc_stmt->publication;

					fc_AlterSubscription_refresh(fc_sub, fc_opts.copy_data,
											  fc_stmt->publication);
				}

				break;
			}

		case ALTER_SUBSCRIPTION_ADD_PUBLICATION:
		case ALTER_SUBSCRIPTION_DROP_PUBLICATION:
			{
				List	   *fc_publist;
				bool		fc_isadd = fc_stmt->kind == ALTER_SUBSCRIPTION_ADD_PUBLICATION;

				fc_supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA;
				fc_parse_subscription_options(fc_pstate, fc_stmt->options,
										   fc_supported_opts, &fc_opts);

				fc_publist = fc_merge_publications(fc_sub->publications, fc_stmt->publication, fc_isadd, fc_stmt->subname);
				fc_values[Anum_pg_subscription_subpublications - 1] =
					fc_publicationListToArray(fc_publist);
				fc_replaces[Anum_pg_subscription_subpublications - 1] = true;

				fc_update_tuple = true;

				/* 如果用户要求我们刷新，则刷新。 */
				if (fc_opts.refresh)
				{
					/* 我们只需要验证用户指定的出版物。 */
					List	   *fc_validate_publications = (fc_isadd) ? fc_stmt->publication : NULL;

					if (!fc_sub->enabled)
						ereport(ERROR,
								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
								 errmsg("ALTER SUBSCRIPTION with refresh is not allowed for disabled subscriptions"),
						/* 翻译者：%s是SQL ALTER命令 */
								 errhint("Use %s instead.",
										 fc_isadd ?
										 "ALTER SUBSCRIPTION ... ADD PUBLICATION ... WITH (refresh = false)" :
										 "ALTER SUBSCRIPTION ... DROP PUBLICATION ... WITH (refresh = false)")));

					/*
					 * 查看ALTER_SUBSCRIPTION_REFRESH以了解为什么
					 * 这不被允许的详细信息。
					 */
					if (fc_sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && fc_opts.copy_data)
						ereport(ERROR,
								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
								 errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
						/* 翻译者：%s是SQL ALTER命令 */
								 errhint("Use %s with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.",
										 fc_isadd ?
										 "ALTER SUBSCRIPTION ... ADD PUBLICATION" :
										 "ALTER SUBSCRIPTION ... DROP PUBLICATION")));

					PreventInTransactionBlock(fc_isTopLevel, "ALTER SUBSCRIPTION with refresh");

					/* 刷新新的出版物列表。 */
					fc_sub->publications = fc_publist;

					fc_AlterSubscription_refresh(fc_sub, fc_opts.copy_data,
											  fc_validate_publications);
				}

				break;
			}

		case ALTER_SUBSCRIPTION_REFRESH:
			{
				if (!fc_sub->enabled)
					ereport(ERROR,
							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
							 errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions")));

				fc_parse_subscription_options(fc_pstate, fc_stmt->options,
										   SUBOPT_COPY_DATA, &fc_opts);

				/*
				 * 订阅选项 "two_phase" 要求在两阶段正式启用之前，
				 * 复制必须通过初始表同步阶段。
				 *
				 * 但是，一旦达到此两阶段提交 "enabled" 状态，
				 * 我们必须不允许任何后续的表初始化发生。因此，
				 * 当用户请求两阶段 = 开模式时，不允许执行 ALTER SUBSCRIPTION ... REFRESH。
				 *
				 * 这一限制的例外是当 copy_data = false，因为当 copy_data 为 false 时，
				 * 表同步将直接以 READY 状态开始，并将直接退出而不做任何事情。
				 *
				 * 有关更多详细信息，请参见 worker.c 顶部的注释。
				 */
				if (fc_sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && fc_opts.copy_data)
					ereport(ERROR,
							(errcode(ERRCODE_SYNTAX_ERROR),
							 errmsg("ALTER SUBSCRIPTION ... REFRESH with copy_data is not allowed when two_phase is enabled"),
							 errhint("Use ALTER SUBSCRIPTION ... REFRESH with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));

				PreventInTransactionBlock(fc_isTopLevel, "ALTER SUBSCRIPTION ... REFRESH");

				fc_AlterSubscription_refresh(fc_sub, fc_opts.copy_data, NULL);

				break;
			}

		case ALTER_SUBSCRIPTION_SKIP:
			{
				fc_parse_subscription_options(fc_pstate, fc_stmt->options, SUBOPT_LSN, &fc_opts);

				/* ALTER SUBSCRIPTION ... SKIP 仅支持 LSN 选项 */
				Assert(IsSet(fc_opts.specified_opts, SUBOPT_LSN));

				if (!superuser())
					ereport(ERROR,
							(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
							 errmsg("must be superuser to skip transaction")));

				/*
				 * 如果用户设置了 subskiplsn，我们会进行有效性检查，以确保
				 * 指定的 LSN 是一个可能的值。
				 */
				if (!XLogRecPtrIsInvalid(fc_opts.lsn))
				{
					RepOriginId fc_originid;
					char		fc_originname[NAMEDATALEN];
					XLogRecPtr	fc_remote_lsn;

					snprintf(fc_originname, sizeof(fc_originname), "pg_%u", fc_subid);
					fc_originid = replorigin_by_name(fc_originname, false);
					fc_remote_lsn = replorigin_get_progress(fc_originid, false);

					/* 检查给定的 LSN 至少是一个未来的 LSN */
					if (!XLogRecPtrIsInvalid(fc_remote_lsn) && fc_opts.lsn < fc_remote_lsn)
						ereport(ERROR,
								(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
								 errmsg("skip WAL location (LSN %X/%X) must be greater than origin LSN %X/%X",
										LSN_FORMAT_ARGS(fc_opts.lsn),
										LSN_FORMAT_ARGS(fc_remote_lsn))));
				}

				fc_values[Anum_pg_subscription_subskiplsn - 1] = LSNGetDatum(fc_opts.lsn);
				fc_replaces[Anum_pg_subscription_subskiplsn - 1] = true;

				fc_update_tuple = true;
				break;
			}

		default:
			elog(ERROR, "unrecognized ALTER SUBSCRIPTION kind %d",
				 fc_stmt->kind);
	}

	/* 如有必要，更新目录。 */
	if (fc_update_tuple)
	{
		fc_tup = heap_modify_tuple(fc_tup, RelationGetDescr(fc_rel), fc_values, fc_nulls,
								fc_replaces);

		CatalogTupleUpdate(fc_rel, &fc_tup->t_self, fc_tup);

		heap_freetuple(fc_tup);
	}

	table_close(fc_rel, RowExclusiveLock);

	ObjectAddressSet(fc_myself, SubscriptionRelationId, fc_subid);

	InvokeObjectPostAlterHook(SubscriptionRelationId, fc_subid, 0);

	return fc_myself;
}

/*
 * 删除订阅
 */
void DropSubscription(DropSubscriptionStmt *fc_stmt, bool fc_isTopLevel)
{
	Relation	fc_rel;
	ObjectAddress fc_myself;
	HeapTuple	fc_tup;
	Oid			fc_subid;
	Datum		fc_datum;
	bool		fc_isnull;
	char	   *fc_subname;
	char	   *fc_conninfo;
	char	   *fc_slotname;
	List	   *fc_subworkers;
	ListCell   *fc_lc;
	char		fc_originname[NAMEDATALEN];
	char	   *fc_err = NULL;
	WalReceiverConn *fc_wrconn;
	Form_pg_subscription fc_form;
	List	   *fc_rstates;

	/*
	 * 用 AccessExclusiveLock 锁定 pg_subscription，以确保
	 * 启动程序在删除订阅期间不会重启新的工作进程。
	 */
	fc_rel = table_open(SubscriptionRelationId, AccessExclusiveLock);

	fc_tup = SearchSysCache2(SUBSCRIPTIONNAME, MyDatabaseId,
						  CStringGetDatum(fc_stmt->subname));

	if (!HeapTupleIsValid(fc_tup))
	{
		table_close(fc_rel, NoLock);

		if (!fc_stmt->missing_ok)
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("subscription \"%s\" does not exist",
							fc_stmt->subname)));
		else
			ereport(NOTICE,
					(errmsg("subscription \"%s\" does not exist, skipping",
							fc_stmt->subname)));

		return;
	}

	fc_form = (Form_pg_subscription) GETSTRUCT(fc_tup);
	fc_subid = fc_form->oid;

	/* 必须是所有者 */
	if (!pg_subscription_ownercheck(fc_subid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
					   fc_stmt->subname);

	/* 正在删除的订阅的 DROP 钩子 */
	InvokeObjectDropHook(SubscriptionRelationId, fc_subid, 0);

	/*
	 * 锁定订阅，以便其他人无法对其进行任何操作（包括
	 * 复制工作进程）。
	 */
	LockSharedObject(SubscriptionRelationId, fc_subid, 0, AccessExclusiveLock);

	/* 获取 subname */
	fc_datum = SysCacheGetAttr(SUBSCRIPTIONOID, fc_tup,
							Anum_pg_subscription_subname, &fc_isnull);
	Assert(!fc_isnull);
	fc_subname = pstrdup(NameStr(*DatumGetName(fc_datum)));

	/* 获取 conninfo */
	fc_datum = SysCacheGetAttr(SUBSCRIPTIONOID, fc_tup,
							Anum_pg_subscription_subconninfo, &fc_isnull);
	Assert(!fc_isnull);
	fc_conninfo = TextDatumGetCString(fc_datum);

	/* 获取 slotname */
	fc_datum = SysCacheGetAttr(SUBSCRIPTIONOID, fc_tup,
							Anum_pg_subscription_subslotname, &fc_isnull);
	if (!fc_isnull)
		fc_slotname = pstrdup(NameStr(*DatumGetName(fc_datum)));
	else
		fc_slotname = NULL;

	/*
	 * 因为删除复制槽并不是事务性的，即使事务回滚，
	 * 复制槽仍然会保持被删除状态。因此，如果删除复制槽，就不能
	 * 在事务块内运行 DROP SUBSCRIPTION。另外，在这种情况下，
	 * 我们会向累积统计系统报告删除订阅的消息。
	 *
	 * XXX 命令名称真的应该类似于“删除与复制槽相关联的订阅的 DROPSUBSCRIPTION”，
	 * 但我们没有适当的设施来实现这一点。
	 */
	if (fc_slotname)
		PreventInTransactionBlock(fc_isTopLevel, "DROP SUBSCRIPTION");

	ObjectAddressSet(fc_myself, SubscriptionRelationId, fc_subid);
	EventTriggerSQLDropAddObject(&fc_myself, true, true);

	/* 从目录中移除元组。 */
	CatalogTupleDelete(fc_rel, &fc_tup->t_self);

	ReleaseSysCache(fc_tup);

	/*
	 * 立即停止所有的订阅工作进程。
	 *
	 * 如果我们正在删除复制槽，这是必要的，以便槽变得可访问。
	 *
	 * 如果订阅被禁用并且在同一事务中被禁用，这也是必要的。
	 * 因此，工作进程尚未看到禁用，仍会运行，这可能导致稍后我们想要
	 * 删除复制来源时出现挂起。如果订阅在这个事务之前被禁用，
	 * 那么就不应该有任何工作进程留下，所以这不会产生差异。
	 *
	 * 新的工作进程不会被启动，因为我们在事务结束之前
	 * 持有对订阅的独占锁。
	 */
	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
	fc_subworkers = logicalrep_workers_find(fc_subid, false);
	LWLockRelease(LogicalRepWorkerLock);
	foreach(fc_lc, fc_subworkers)
	{
		LogicalRepWorker *fc_w = (LogicalRepWorker *) lfirst(fc_lc);

		logicalrep_worker_stop(fc_w->subid, fc_w->relid);
	}
	list_free(fc_subworkers);

	/*
	 * 清理 tablesync 复制源。
	 *
	 * 任何 READY 状态的关系已经处理过清理。
	 *
	 * 注意状态不能改变，因为我们已经停止了应用和 tablesync 工作线程，并且由于订阅上的独占锁，它们无法重新启动。
	 */
	fc_rstates = GetSubscriptionNotReadyRelations(fc_subid);
	foreach(fc_lc, fc_rstates)
	{
		SubscriptionRelState *fc_rstate = (SubscriptionRelState *) lfirst(fc_lc);
		Oid			fc_relid = fc_rstate->relid;

		/* 仅清理 tablesync 工作线程的资源 */
		if (!OidIsValid(fc_relid))
			continue;

		/*
		 * 如果存在，删除 tablesync 的源跟踪。
		 *
		 * 可能 tablesync 工作线程尚未创建源，因此传递 missing_ok = true。这可能发生在 SUBREL_STATE_FINISHEDCOPY 之前的状态。
		 */
		ReplicationOriginNameForTablesync(fc_subid, fc_relid, fc_originname,
										  sizeof(fc_originname));
		replorigin_drop_by_name(fc_originname, true, false);
	}

	/* 清理依赖关系 */
	deleteSharedDependencyRecordsFor(SubscriptionRelationId, fc_subid, 0);

	/* 移除任何关联的关系同步状态。 */
	RemoveSubscriptionRel(fc_subid, InvalidOid);

	/* 如果存在，移除源跟踪。 */
	snprintf(fc_originname, sizeof(fc_originname), "pg_%u", fc_subid);
	replorigin_drop_by_name(fc_originname, true, false);

	/*
	 * 告诉累积统计系统订阅正在被删除。
	 */
	pgstat_drop_subscription(fc_subid);

	/*
	 * 如果没有与订阅相关联的槽，我们可以在这里结束。
	 */
	if (!fc_slotname && fc_rstates == NIL)
	{
		table_close(fc_rel, NoLock);
		return;
	}

	/*
	 * 尝试获取删除槽所需的连接。
	 *
	 * 注意：如果槽名称为 NONE/NULL，则允许命令完成，用户需要手动清理应用和 tablesync 工作线程的槽。
	 *
	 * 这必须放在最后，因为如果在进行数据库操作时发生错误，我们将无法回滚已删除的槽。
	 */
	load_file("libpqwalreceiver", false);

	fc_wrconn = walrcv_connect(fc_conninfo, true, fc_subname, &fc_err);
	if (fc_wrconn == NULL)
	{
		if (!fc_slotname)
		{
			/* 保持整洁 */
			list_free(fc_rstates);
			table_close(fc_rel, NoLock);
			return;
		}
		else
		{
			fc_ReportSlotConnectionError(fc_rstates, fc_subid, fc_slotname, fc_err);
		}
	}

	PG_TRY();
	{
		foreach(fc_lc, fc_rstates)
		{
			SubscriptionRelState *fc_rstate = (SubscriptionRelState *) lfirst(fc_lc);
			Oid			fc_relid = fc_rstate->relid;

			/* 仅清理 tablesync 工作线程的资源 */
			if (!OidIsValid(fc_relid))
				continue;

			/*
			 * 删除与已移除表关联的 tablesync 槽。
			 *
			 * 对于 SYNCDONE/READY 状态，已知 tablesync 槽已经由 tablesync 工作线程删除。
			 *
			 * 对于其他状态，并不能确定，槽可能还不存在。此外，如果在删除了一些槽后失败，下次将再次尝试删除已经删除的槽并失败。出于这些原因，我们允许去掉时 missing_ok = true。
			 */
			if (fc_rstate->state != SUBREL_STATE_SYNCDONE)
			{
				char		fc_syncslotname[NAMEDATALEN] = {0};

				ReplicationSlotNameForTablesync(fc_subid, fc_relid, fc_syncslotname,
												sizeof(fc_syncslotname));
				ReplicationSlotDropAtPubNode(fc_wrconn, fc_syncslotname, true);
			}
		}

		list_free(fc_rstates);

		/*
		 * 如果与订阅关联的槽存在，则在发布者处删除复制槽。
		 */
		if (fc_slotname)
			ReplicationSlotDropAtPubNode(fc_wrconn, fc_slotname, false);
	}
	PG_FINALLY();
	{
		walrcv_disconnect(fc_wrconn);
	}
	PG_END_TRY();

	table_close(fc_rel, NoLock);
}

/*
 * 使用复制连接在发布者节点删除复制槽。
 *
 * missing_ok - 如果为 true，则仅在槽不存在时发出一个日志消息。
 */
void ReplicationSlotDropAtPubNode(WalReceiverConn *fc_wrconn, char *fc_slotname, bool fc_missing_ok)
{
	StringInfoData fc_cmd;

	Assert(fc_wrconn);

	load_file("libpqwalreceiver", false);

	initStringInfo(&fc_cmd);
	appendStringInfo(&fc_cmd, "DROP_REPLICATION_SLOT %s WAIT", quote_identifier(fc_slotname));

	PG_TRY();
	{
		WalRcvExecResult *fc_res;

		fc_res = walrcv_exec(fc_wrconn, fc_cmd.data, 0, NULL);

		if (fc_res->status == WALRCV_OK_COMMAND)
		{
			/* 注意。成功。 */
			ereport(NOTICE,
					(errmsg("dropped replication slot \"%s\" on publisher",
							fc_slotname)));
		}
		else if (fc_res->status == WALRCV_ERROR &&
				 fc_missing_ok &&
				 fc_res->sqlstate == ERRCODE_UNDEFINED_OBJECT)
		{
			/* 日志。错误，但 missing_ok = true。 */
			ereport(LOG,
					(errmsg("could not drop replication slot \"%s\" on publisher: %s",
							fc_slotname, fc_res->err)));
		}
		else
		{
			/* 错误。 */
			ereport(ERROR,
					(errcode(ERRCODE_CONNECTION_FAILURE),
					 errmsg("could not drop replication slot \"%s\" on publisher: %s",
							fc_slotname, fc_res->err)));
		}

		walrcv_clear_result(fc_res);
	}
	PG_FINALLY();
	{
		pfree(fc_cmd.data);
	}
	PG_END_TRY();
}

/*
 * 内部工作程序用于更改订阅拥有者
 */
static void fc_AlterSubscriptionOwner_internal(Relation fc_rel, HeapTuple fc_tup, Oid fc_newOwnerId)
{
	Form_pg_subscription fc_form;

	fc_form = (Form_pg_subscription) GETSTRUCT(fc_tup);

	if (fc_form->subowner == fc_newOwnerId)
		return;

	if (!pg_subscription_ownercheck(fc_form->oid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
					   NameStr(fc_form->subname));

	/* 新所有者必须是超级用户 */
	if (!superuser_arg(fc_newOwnerId))
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("permission denied to change owner of subscription \"%s\"",
						NameStr(fc_form->subname)),
				 errhint("The owner of a subscription must be a superuser.")));

	fc_form->subowner = fc_newOwnerId;
	CatalogTupleUpdate(fc_rel, &fc_tup->t_self, fc_tup);

	/* 更新所有者依赖关系引用 */
	changeDependencyOnOwner(SubscriptionRelationId,
							fc_form->oid,
							fc_newOwnerId);

	InvokeObjectPostAlterHook(SubscriptionRelationId,
							  fc_form->oid, 0);

	ApplyLauncherWakeupAtCommit();
}

/*
 * 按名称更改订阅拥有者
 */
ObjectAddress AlterSubscriptionOwner(const char *fc_name, Oid fc_newOwnerId)
{
	Oid			fc_subid;
	HeapTuple	fc_tup;
	Relation	fc_rel;
	ObjectAddress fc_address;
	Form_pg_subscription fc_form;

	fc_rel = table_open(SubscriptionRelationId, RowExclusiveLock);

	fc_tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, MyDatabaseId,
							  CStringGetDatum(fc_name));

	if (!HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("subscription \"%s\" does not exist", fc_name)));

	fc_form = (Form_pg_subscription) GETSTRUCT(fc_tup);
	fc_subid = fc_form->oid;

	fc_AlterSubscriptionOwner_internal(fc_rel, fc_tup, fc_newOwnerId);

	ObjectAddressSet(fc_address, SubscriptionRelationId, fc_subid);

	heap_freetuple(fc_tup);

	table_close(fc_rel, RowExclusiveLock);

	return fc_address;
}

/*
 * 按 OID 更改订阅拥有者
 */
void AlterSubscriptionOwner_oid(Oid fc_subid, Oid fc_newOwnerId)
{
	HeapTuple	fc_tup;
	Relation	fc_rel;

	fc_rel = table_open(SubscriptionRelationId, RowExclusiveLock);

	fc_tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(fc_subid));

	if (!HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("subscription with OID %u does not exist", fc_subid)));

	fc_AlterSubscriptionOwner_internal(fc_rel, fc_tup, fc_newOwnerId);

	heap_freetuple(fc_tup);

	table_close(fc_rel, RowExclusiveLock);
}

/*
 * 获取属于指定发布的表的列表，位于发布者连接上。
 *
 * 请注意，我们不支持在不同发布中同一表的列列表不同的情况，以避免为某些行发送不必要的列信息。当为不同的发布指定列列表和行过滤器时，可能会发生这种情况。
 */
static List * fc_fetch_table_list(WalReceiverConn *fc_wrconn, List *fc_publications)
{
	WalRcvExecResult *fc_res;
	StringInfoData fc_cmd;
	TupleTableSlot *fc_slot;
	Oid			fc_tableRow[3] = {TEXTOID, TEXTOID, NAMEARRAYOID};
	List	   *fc_tablelist = NIL;
	bool		fc_check_columnlist = (walrcv_server_version(fc_wrconn) >= 150000);

	initStringInfo(&fc_cmd);
	appendStringInfoString(&fc_cmd, "SELECT DISTINCT t.schemaname, t.tablename \n");

	/* 如果发布者支持，获取每个关系的列列表 */
	if (fc_check_columnlist)
		appendStringInfoString(&fc_cmd, ", t.attnames\n");

	appendStringInfoString(&fc_cmd, "FROM pg_catalog.pg_publication_tables t\n"
						   " WHERE t.pubname IN (");
	fc_get_publications_str(fc_publications, &fc_cmd, true);
	appendStringInfoChar(&fc_cmd, ')');

	fc_res = walrcv_exec(fc_wrconn, fc_cmd.data, fc_check_columnlist ? 3 : 2, fc_tableRow);
	pfree(fc_cmd.data);

	if (fc_res->status != WALRCV_OK_TUPLES)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("could not receive list of replicated tables from the publisher: %s",
						fc_res->err)));

	/* 处理表。 */
	fc_slot = MakeSingleTupleTableSlot(fc_res->tupledesc, &TTSOpsMinimalTuple);
	while (tuplestore_gettupleslot(fc_res->tuplestore, true, false, fc_slot))
	{
		char	   *fc_nspname;
		char	   *fc_relname;
		bool		fc_isnull;
		RangeVar   *fc_rv;

		fc_nspname = TextDatumGetCString(slot_getattr(fc_slot, 1, &fc_isnull));
		Assert(!fc_isnull);
		fc_relname = TextDatumGetCString(slot_getattr(fc_slot, 2, &fc_isnull));
		Assert(!fc_isnull);

		fc_rv = makeRangeVar(fc_nspname, fc_relname, -1);

		if (fc_check_columnlist && list_member(fc_tablelist, fc_rv))
			ereport(ERROR,
					errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					errmsg("cannot use different column lists for table \"%s.%s\" in different publications",
						   fc_nspname, fc_relname));
		else
			fc_tablelist = lappend(fc_tablelist, fc_rv);

		ExecClearTuple(fc_slot);
	}
	ExecDropSingleTupleTableSlot(fc_slot);

	walrcv_clear_result(fc_res);

	return fc_tablelist;
}

/*
 * 这是在删除复制槽时报告连接失败。
 * 在这里，我们对所有的表同步槽报告警告，以便用户在需要时手动删除它们。
 */
static void fc_ReportSlotConnectionError(List *fc_rstates, Oid fc_subid, char *fc_slotname, char *fc_err)
{
	ListCell   *fc_lc;

	foreach(fc_lc, fc_rstates)
	{
		SubscriptionRelState *fc_rstate = (SubscriptionRelState *) lfirst(fc_lc);
		Oid			fc_relid = fc_rstate->relid;

		/* 仅清理 tablesync 工作线程的资源 */
		if (!OidIsValid(fc_relid))
			continue;

		/*
		 * 调用者需要确保 relstate 在我们之下不会改变。
		 * 请参阅 DropSubscription，我们在其中获取 relstates。
		 */
		if (fc_rstate->state != SUBREL_STATE_SYNCDONE)
		{
			char		fc_syncslotname[NAMEDATALEN] = {0};

			ReplicationSlotNameForTablesync(fc_subid, fc_relid, fc_syncslotname,
											sizeof(fc_syncslotname));
			elog(WARNING, "could not drop tablesync replication slot \"%s\"",
				 fc_syncslotname);
		}
	}

	ereport(ERROR,
			(errcode(ERRCODE_CONNECTION_FAILURE),
			 errmsg("could not connect to publisher when attempting to drop replication slot \"%s\": %s",
					fc_slotname, fc_err),
	/* 翻译者：%s是SQL ALTER命令 */
			 errhint("Use %s to disable the subscription, and then use %s to disassociate it from the slot.",
					 "ALTER SUBSCRIPTION ... DISABLE",
					 "ALTER SUBSCRIPTION ... SET (slot_name = NONE)")));
}

/*
 * 检查给定发布列表中的重复项，如果找到则报错。
 * 如果 datums 不为 NULL，则将发布添加到 datums 作为文本数据项。
 */
static void fc_check_duplicates_in_publist(List *fc_publist, Datum *fc_datums)
{
	ListCell   *fc_cell;
	int			fc_j = 0;

	foreach(fc_cell, fc_publist)
	{
		char	   *fc_name = strVal(lfirst(fc_cell));
		ListCell   *fc_pcell;

		foreach(fc_pcell, fc_publist)
		{
			char	   *fc_pname = strVal(lfirst(fc_pcell));

			if (fc_pcell == fc_cell)
				break;

			if (strcmp(fc_name, fc_pname) == 0)
				ereport(ERROR,
						(errcode(ERRCODE_DUPLICATE_OBJECT),
						 errmsg("publication name \"%s\" used more than once",
								fc_pname)));
		}

		if (fc_datums)
			fc_datums[fc_j++] = CStringGetTextDatum(fc_name);
	}
}

/*
 * 合并当前订阅的发布和用户指定的出版物
 * 来自 ADD/DROP PUBLICATIONS。
 *
 * 如果 addpub 为真，我们将把发布列表添加到 oldpublist。
 * 否则，我们将从 oldpublist 中删除发布列表。返回的列表是一个副本，oldpublist 本身没有更改。
 *
 * subname 是订阅名称，用于错误消息。
 */
static List * fc_merge_publications(List *fc_oldpublist, List *fc_newpublist, bool fc_addpub, const char *fc_subname)
{
	ListCell   *fc_lc;

	fc_oldpublist = list_copy(fc_oldpublist);

	fc_check_duplicates_in_publist(fc_newpublist, NULL);

	foreach(fc_lc, fc_newpublist)
	{
		char	   *fc_name = strVal(lfirst(fc_lc));
		ListCell   *fc_lc2;
		bool		fc_found = false;

		foreach(fc_lc2, fc_oldpublist)
		{
			char	   *fc_pubname = strVal(lfirst(fc_lc2));

			if (strcmp(fc_name, fc_pubname) == 0)
			{
				fc_found = true;
				if (fc_addpub)
					ereport(ERROR,
							(errcode(ERRCODE_DUPLICATE_OBJECT),
							 errmsg("publication \"%s\" is already in subscription \"%s\"",
									fc_name, fc_subname)));
				else
					fc_oldpublist = foreach_delete_current(fc_oldpublist, fc_lc2);

				break;
			}
		}

		if (fc_addpub && !fc_found)
			fc_oldpublist = lappend(fc_oldpublist, makeString(fc_name));
		else if (!fc_addpub && !fc_found)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
					 errmsg("publication \"%s\" is not in subscription \"%s\"",
							fc_name, fc_subname)));
	}

	/*
	 * XXX 可能没有强有力的理由这样做，但目前这是为了使 ALTER
	 * SUBSCRIPTION ... DROP PUBLICATION 与 SET PUBLICATION 一致。
	 */
	if (!fc_oldpublist)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
				 errmsg("cannot drop all the publications from a subscription")));

	return fc_oldpublist;
}
